package com.zhang.hadoop.flink.test9;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: zhang yufei
 * @createTime:2022/9/24 16:10
 * @description:
 */
public class CommonApiTest {

    public static void main(String[] args) throws Exception {

        //1.定义环境配置来创建表执行环境
        //基于flink版本planner进行流处理
        EnvironmentSettings setting = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(setting);

        //2.创建表
        String createDDL="create table clickTable("+
                " user_name string,"+
                " url string,"+
                " ts bigint"+
                ") with ("+
                " 'connector' = 'filesystem',"+
                " 'path'='flink/input/clicks',"+
                " 'format'='csv'"+
                ")";

        tableEnv.executeSql(createDDL);

        Table result = tableEnv.from("clickTable");

        tableEnv.createTemporaryView("result", result);

        Table resultTable=tableEnv.sqlQuery("select url,user_name from clickTable");

        //创建一张用于输出的表
        String createOutDDL="create table outTable("+
                " url string,"+
                " user_name string"+
                ") with ("+
                " 'connector' = 'filesystem',"+
                " 'path'='flink/output/clicks',"+
                " 'format'='csv'"+
                ")";

        tableEnv.executeSql(createOutDDL);

        //输出表
        resultTable.executeInsert("outTable");

        //创建一张用于控制台打印
        String createPrintOutDDL="create table printOutTable("+
                " url string,"+
                " user_name string"+
                ") with ("+
                " 'connector' = 'print'"+
                ")";
        tableEnv.executeSql(createPrintOutDDL);
        //输出表
        resultTable.executeInsert("printOutTable");
    }
}
